//package org.lk.job;
//
//import com.ververica.cdc.connectors.oracle.OracleSource;
//import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
//import org.apache.flink.core.fs.Path;
//import org.apache.flink.runtime.state.filesystem.FsStateBackend;
//import org.apache.flink.streaming.api.CheckpointingMode;
//import org.apache.flink.streaming.api.environment.CheckpointConfig;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.streaming.api.functions.source.SourceFunction;
//import org.apache.flink.streaming.connectors.oraclecdc.OracleCDCSource;
//import org.apache.flink.streaming.connectors.oraclecdc.debezium.DebeziumDeserializationSchema;
//
//public class FlinkCDCSource {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        // 设置 checkpoint 的配置
//        env.enableCheckpointing(60000);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.setStateBackend(new FsStateBackend(new Path("http://172.16.8.200:9870/checkpoint/")));
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
//        env.getCheckpointConfig().setCheckpointTimeout(10000);
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//
//        // 读取配置文件中的检查点位置
//        String checkpointPath = "http://172.16.8.200:9870/checkpoint/";
//        if (env.getCheckpointConfig().isCheckpointingEnabled() && env.getCheckpointConfig().getCheckpointStorage() instanceof FsStateBackend) {
//            checkpointPath = ((FsStateBackend) env.getCheckpointConfig().getCheckpointStorage()).getBasePath().toString();
//        }
//
//        // 启动 Oracle CDC 任务
//        startOracleCDCSource(env, "172.16.11.193", 1521, "ocean_config", "ocean_config", "your-database-name", "your-table-name", checkpointPath);
//
//        env.execute("Flink CDC Oracle Source");
//    }
//
//    private static void startOracleCDCSource(StreamExecutionEnvironment env, String hostname, int port, String username, String password, String databaseName, String tableName, String checkpointPath) {
//        OracleSource.Builder<String> deserializer = OracleSource.<String>builder()
//                .hostname("172.16.11.193")
//                .port(1521)
//                .database("XE") // monitor XE database
//                .tableList("inventory.products") // monitor products table
//                .username("flinkuser")
//                .password("flinkpw")
//                .deserializer(new JsonDebeziumDeserializationSchema());
//        // 设置检查点位置
//        if (checkpointPath != null && !checkpointPath.isEmpty()) {
////            deserializer.checkpointRestorePath(new Path(checkpointPath));
//        }
//
//        env.addSource(deserializer.build())
//                .print();
//    }
//}
